package cn.itcast.xc.analysis.visit

import cn.itcast.xc.common.EtlEnvironment
import cn.itcast.xc.entity.{AreaDimen, CourseDimen, CourseVisitFact}
import cn.itcast.xc.utils.DateUtils.getDateStr
import cn.itcast.xc.utils.IpUtils.getArea
import cn.itcast.xc.utils.StrUtils.getUuId

/**
 * <p>
 * 课程访问量， 事实表清洗程序
 * 1，关联维度
 * 2，过滤机器人
 * 3，去重
 * </p>
 **/
object VisitSourceToFact {

  /**
   * 初始化spark session
   */
  val spark = EtlEnvironment.getSparkSession(this.getClass.getSimpleName)

  def main(args: Array[String]): Unit = {

    import spark.implicits._

    // 获取 spark context
    val sc = spark.sparkContext

//    val data_info = "2019-11-18"
    val data_info = args(0)

    // 获取维度数据
    // 区域维度
    var areaMap = Map[String, String]()
    spark.sql("SELECT * FROM data_dimen.area_dim").as[AreaDimen].collect().map(x => {
      areaMap += (x.city_name -> x.area_dim_id.toString)
    })
    // 公司维度
    var courseMap = Map[String, String]()
    spark.sql("SELECT * FROM data_dimen.course_dim").as[CourseDimen].collect().map(x => {
      if (x.course_dim_id != null && x.company_id != null) {
        courseMap += (x.course_dim_id.toString -> x.company_id.toString)
      }
    })

    // 访问量原始数据
    val visitSource = sc.textFile(s"/user/hive/external/data_course/course_visit_source/${data_info}")
    // 进行清洗
    val visitClean = visitSource
      // 分割得到每个字段
      .map(_.split("\t"))
      //过滤机器人
      .filter(arr => {
        !arr(5).contains("Rebot")
      })
      // 批量生产数据进行清洗时，以date_info为准清，需要过滤日期
      // 测试时， 这个一般不过滤， 清洗所有方便报表系统使用
      //      .filter(arr => {
      //        data_info == getDateStr(arr(6).toLong * 1000, "yyyy-MM-dd")
      //      })
      // 关联维度， 转换为事实表的实体对象
      .map(tmp => {
        // 时间维度
        val time_dim_id = getDateStr(tmp(6).toLong * 1000, "yyyy-MM-dd-HH")
        // 区域
        var areaId = "-1"
        try {
          areaId = areaMap.get(getArea(tmp(3))).get
        } catch {
          case e =>
        }
        // 公司
        var comId = "-1"
        try {
          comId = courseMap.get(tmp(0)).get
        } catch {
          case e =>
        }
        // 年月日分区
        val ymd = time_dim_id.split("-")
        // 转换为对象
        CourseVisitFact(getUuId(), tmp(0), time_dim_id, areaId, comId, tmp(3), tmp(2), tmp(5), tmp(6).toLong * 1000, ymd(0), ymd(1), ymd(2))
      })
      // 根据session和ip, 课程id, 选择去重
      .groupBy(x =>
        x.client_ip + x.client_sessionid + x.course_dim_id
      )
      // 重复的值取时间小的
      .map {
        x =>
          x._2.minBy {
            _.visit_time
          }
      }

    // 转换为df
    val visitDF = spark.createDataFrame(visitClean)
    // 配置允许动态分区
    spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict")
    // 保存到数据仓库
    visitDF.repartition(1)
      .write.mode("overwrite")
      .insertInto("data_course.course_visit_fact")
    // 查看结果
    spark.sql("select * from data_course.course_visit_fact").show()

    // 关闭资源
    sc.stop()
    spark.close()
  }

}
